package com.atguigu.app

import java.util

import com.alibaba.fastjson.JSON
import com.atguigu.bean.{OrderDetail, OrderInfo, SaleDetail, UserInfo}
import com.atguigu.constants.GmallConstants
import com.atguigu.utils.{MyEsUtil, MyKafkaUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.json4s.native.Serialization
import redis.clients.jedis.Jedis

import collection.JavaConverters._

object SaleDetailApp {
    def main(args: Array[String]): Unit = {
      //1.创建sparkConf
      val sparkConf: SparkConf = new SparkConf().setAppName("SaleDetailApp").setMaster("local[*]")

      //2.创建StreamingContext
      val ssc: StreamingContext = new StreamingContext(sparkConf, Seconds(3))

      //3.消费kafka数据
      val orderkafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER, ssc)
      val orderDetailkafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER_DETAIL, ssc)

      //4.将数据转为样例类，并补全字段
      val orderInfoDStream = orderkafkaDStream.mapPartitions(partition => {
        partition.map(record => {
          val orderInfo: OrderInfo = JSON.parseObject(record.value(), classOf[OrderInfo])

          //补全字段
          orderInfo.create_date = orderInfo.create_time.split(" ")(0)
          orderInfo.create_hour = orderInfo.create_time.split(" ")(1).split(":")(0)

          (orderInfo.id,orderInfo)
        })
      })

      val orderDetailDStream = orderDetailkafkaDStream.mapPartitions(partition => {
        partition.map(record => {
          val orderDetail: OrderDetail = JSON.parseObject(record.value(), classOf[OrderDetail])
          (orderDetail.order_id,orderDetail)
        })
      })

      //5.两条流做join操作
//      val joinDStream: DStream[(String, (OrderInfo, OrderDetail))] = orderInfoDStream.join(orderDetailDStream)
      val joinDStream: DStream[(String, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoDStream.fullOuterJoin(orderDetailDStream)

      //6.采用加缓存的方式处理因网络延迟所带来的数据丢失问题
      val noUserSaleDetailDStream: DStream[SaleDetail] = joinDStream.mapPartitions(patition => {
        //创建Redis连接
        val jedis: Jedis = new Jedis("hadoop102", 6379)
        implicit val formats = org.json4s.DefaultFormats
        //创建List集合用来存放SaleDetail样例类
        val details: util.ArrayList[SaleDetail] = new util.ArrayList[SaleDetail]()

        patition.foreach { case (orderId, (infoOpt, detailOpt)) =>
          val orderInfoRedisKey: String = "orderInfo:" + orderId
          val orderDetailRedisKey: String = "orderDetail:" + orderId

          //a.判断订单表是否存在
          if (infoOpt.isDefined) {
            //订单表数据存在
            //取出订单表数据
            val orderInfo: OrderInfo = infoOpt.get
            //a.2判断订单明细表是否存在
            if (detailOpt.isDefined) {
              //订单明细表在
              //取出订单明细表数据
              val orderDetail: OrderDetail = detailOpt.get
              //a.3同批次能够关联上，组成SaleDetail样例类关联两张表的数据
              val detail: SaleDetail = new SaleDetail(orderInfo, orderDetail)
              details.add(detail)
            }
            //b.将orderInfo数据写入Redis
            //b.1将样例类转为Json字符串
            //编译报错
            //            JSON.toJSONString(orderInfo)
            val orderInfoJson: String = Serialization.write(orderInfo)
            //b.2将转化后的Json字符串写入Redis中
            jedis.set(orderInfoRedisKey, orderInfoJson)
            //b.3将存入的OrderInfo数据设置过期时间
            jedis.expire(orderInfoRedisKey, 30)

            //c.查询orderDetail缓存中是否有能够关联上的数据
            //c.1首先判断能够关联上的数据的key是否存在
            if (jedis.exists(orderDetailRedisKey)) {
              //redis中有能够关联上的数据
              //c.2将能够关联上的orderDetail数据查询出来
              val orderDetailJsonSet: util.Set[String] = jedis.smembers(orderDetailRedisKey)
              //c.3遍历查询出来的集合，获取到每一条能够与当前orderInfo关联上的数据
              for (elem <- orderDetailJsonSet.asScala) {
                //c.4将遍历出来的orderDetail的json字符串转为样例类
                val orderDetail: OrderDetail = JSON.parseObject(elem, classOf[OrderDetail])
                //c.5将orderInfo与查询出来的orderDetail数据关联起来组成SaleDetail
                val detail: SaleDetail = new SaleDetail(orderInfo, orderDetail)
                details.add(detail)
              }
            }
          } else {
            //orderInfo不在
            //d.判断detailOpt中orderDetail是否存在
            if (detailOpt.isDefined) {
              //orderDetail存在
              //取出orderDetial数据
              val orderDetail: OrderDetail = detailOpt.get
              //d.2去orderInfo缓存中查询是否有能够关联上的数据
              //首先判断对应orderInfo的RedisKey是否存在
              if (jedis.exists(orderInfoRedisKey)) {
                //存在这个key，就证明有能够关联上的OrderInfo数据
                //d.3取出对应的OrderInfo数据
                val orderInfoJsonStr: String = jedis.get(orderInfoRedisKey)
                //d.4将其转为样例类
                val orderInfo: OrderInfo = JSON.parseObject(orderInfoJsonStr, classOf[OrderInfo])
                //d.5关联数据，组成SaleDetail样例类
                val detail: SaleDetail = new SaleDetail(orderInfo, orderDetail)
                details.add(detail)
              } else {
                //orderInfoRediskey不存在，则证明缓存中没有能够关联上的OrderInfo数据
                //e.将orderDetail自己存入缓存
                //e.1将orderDetail样例类转为JSON字符串
                val orderDetailJsonStr: String = Serialization.write(orderDetail)
                //e.2将转化后的json字符串写入Redis
                jedis.sadd(orderDetailRedisKey, orderDetailJsonStr)
                //e.3设置过期时间
                jedis.expire(orderDetailRedisKey, 30)
              }
            }
          }
        }
        jedis.close()
        //将存放SaleDetail样例类的结果集合当做是mapPartition的返回值
        //先将java集合转为Scala集合然后在转为迭代器（因为mapPartition需要的是一个Scala的迭代器作为返回值）
        details.asScala.toIterator
      })
//      noUserSaleDetailDStream.print()

      //7.反查Redis缓存关联用户表数据
      val saleDetailDStream: DStream[SaleDetail] = noUserSaleDetailDStream.mapPartitions(partition => {
        //创建Redis链接
        val jedis: Jedis = new Jedis("hadoop102", 6379)
        val details: Iterator[SaleDetail] = partition.map(saleDetail => {
          //查询Reids中用户表的数据，并关联
          val userInfoRedisKey: String = "userInfo:" + saleDetail.user_id
          val userInfoJsonStr: String = jedis.get(userInfoRedisKey)
          //将查询出来的Json字符串转为样例类
          val userInfo: UserInfo = JSON.parseObject(userInfoJsonStr, classOf[UserInfo])
          saleDetail.mergeUserInfo(userInfo)
          saleDetail
        })
        jedis.close()
        details
      })
      saleDetailDStream.print()

      //8.将关联后的明细数据写入ES
      saleDetailDStream.foreachRDD(rdd=>{
        rdd.foreachPartition(partition=>{
          val list: List[(String, SaleDetail)] = partition.toList.map(saleDetail => {
            (saleDetail.order_detail_id, saleDetail)
          })
         MyEsUtil.insertBulk(GmallConstants.ES_INDEX_SALEDETAIL+"220212",list)
        })
      })



      ssc.start()
      ssc.awaitTermination()
    }
}
